package cn.tedu.charging.order.consumer;

import cn.tedu.charging.common.consts.RabbitConstants;
import cn.tedu.charging.common.enums.GunStatusEnum;
import cn.tedu.charging.common.message.DelayMsg;
import cn.tedu.charging.common.param.GunStatusUpdateParam;
import cn.tedu.charging.common.protocol.JsonResult;
import cn.tedu.charging.common.vo.WebSocketMessage;
import cn.tedu.charging.order.feign.DeviceClient;
import cn.tedu.charging.order.po.BillFailPO;
import cn.tedu.charging.order.repository.BillFailRepository;
import cn.tedu.charging.order.repository.BillSuccessRepository;
import cn.tedu.charging.order.server.WebSocketServer;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * 延迟消费的消费者
 */
@Component
@Slf4j
public class OrderDelayConsumer {
    @Autowired
    private BillSuccessRepository billSuccessRepository;
    @Autowired
    private BillFailRepository billFailRepository;
    @Autowired
    private DeviceClient client;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private WebSocketServer webSocketServer;
    /**
     * 接收延迟生产消息的数据DelayMsg
     */
    @RabbitListener(queues = RabbitConstants.DELAY_CHARGING_QUEUE)
    public void delayCheckMsgConsume(DelayMsg msg){
        log.info("延迟消息接收到,msg:{}",msg);
        //抢锁setnx是String类型数据 获取操作String类型的对象
        ValueOperations<String,String> opsForValue = redisTemplate.opsForValue();
        //1.调用一个幂等的方法 处理这个消息数据

            //抢锁 设计抢锁key值
            String lockKey="charging:delay:msg:"+msg.getOrderNo();
            boolean success=false;
            //抢锁次数判断
            Integer count=0;
            do{
                //第一次还是后续几次
                if (count>0){
                    log.info("该线程:{},抢锁:{},失败,等待1秒钟 再次抢锁",Thread.currentThread().getId(),lockKey);
                    //说明不是第一次 等待1秒钟再强
                    try{
                        Thread.sleep(1000);
                    }catch (InterruptedException e){
                        log.error("线程被破坏了");
                    }
                }
                success=opsForValue.setIfAbsent(lockKey, "1",3, TimeUnit.SECONDS);
                //抢了几次 当前线程的count就是几
                count++;
            }while (!success);
            //抢到锁的线程才能执行业务方法
            try{
                handleDelayResult(msg);
                //不能在业务方法下面 同步执行释放锁
            }catch (Exception e){
                log.error("执行业务逻辑异常",e);
            }finally {
                //释放资源一定放到finally
                redisTemplate.delete(lockKey);
            }

//            Boolean success = opsForValue.setIfAbsent(lockKey, "1");//setnx lockKey 1
//            //判断 只有成功的succes线程才能执行业务方法其他都结束
//            if(success){
//                handleDelayResult(msg);//业务核心方法
//            }else{
//                log.error("该线程:{},抢锁:{},失败,等待1秒钟 再次抢锁",Thread.currentThread().getId(),lockKey);
//            }

    }
    //synchronized不仅仅解决不了分布式环境的 并发安全问题 还增加了程序负担导致性能下降
    private void handleDelayResult(DelayMsg msg) {
        //1.拿到订单编号
        String billId=msg.getOrderNo();
        //2.检查成功和失败的订单是否存在
        boolean exists=false;
        boolean exists1=checkFailExists(billId);
        log.info("延迟消费发现订单:{},在失败存在结果:{}",billId,exists);
        boolean exists2=checkSuccessExists(billId);
        exists=exists1||exists2;
        log.info("延迟消费发现订单:{},在成功存在结果:{}",billId,exists);
        if (exists){
            //不能再新增了 已经有反馈了
            log.error("该订单:{},已经有处理结果,不能再次处理",billId);
            return;
        }else{
            //新增bill
            log.info("该订单:{},没有处理结果,说明设备严重超时",billId);
            BillFailPO po=new BillFailPO();
            po.setBillId(billId);
            //其他信息 gunId stationId pileId
            po.setCreateTime(new Date());
            po.setUpdateTime(new Date());
            po.setUserId(msg.getUserId());
            po.setFailDesc("设备反馈超时");
            po.setGunId(msg.getGunId());
            billFailRepository.save(po);
            //更改枪状态
            GunStatusUpdateParam param=new GunStatusUpdateParam();
            param.setGunId(msg.getGunId());
            param.setStatus(GunStatusEnum.FAULT.getStatus());
            client.updateGunStatus(param);
            //发送给用户 当前枪有问题
            log.info("当前处理订单的userId:{}",msg.getUserId());
            WebSocketMessage<String> message=new WebSocketMessage<>();
            message.setState(1);
            message.setData("B对不起 充电桩故障 请换枪");
            String jsonString = JSON.toJSONString(message);
            webSocketServer.pushMessage(jsonString,msg.getUserId());
            log.info("反馈给前端 信息:{}",jsonString);
        }
    }

    private boolean checkSuccessExists(String billId) {
        QueryWrapper queryWrapper=new QueryWrapper();
        queryWrapper.eq("bill_id",billId);
        long count = billSuccessRepository.count(queryWrapper);
        return count>0;
    }

    private boolean checkFailExists(String billId) {
        QueryWrapper queryWrapper=new QueryWrapper();
        queryWrapper.eq("bill_id",billId);
        long count = billFailRepository.count(queryWrapper);
        return count>0;
    }
}
